package com.neo.agent.connection;

import com.alibaba.fastjson.JSON;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.Filters;
import com.neo.push.core.RegisterMessage;
import com.neo.push.core.connection.ConnectionListener;
import com.neo.push.core.connection.SocketConnection;
import com.neo.push.server.mongo.MongoService;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;

/**
 * Created by lock on 17-8-1.
 */
public class MongoConnectionListener implements ConnectionListener,ConnectionService {


    private MongoService mongoService;
    private String collectionName;
    private MongoCollection<Document> messageCollection;
    private ConnectionRenderListener registerListener;
    private ConnectionRenderListener closedListener;
    private Logger logger = LoggerFactory.getLogger(MongoConnectionListener.class);

    public void init(){
        try {
            messageCollection = mongoService.getCollection(collectionName);
            messageCollection.find().forEach((Consumer<Document>) document -> messageCollection.deleteOne(document));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 将当前channelId与连接信息绑定,或者去做校验，等等
     * @param connection            哪个连接
     * @param registerMessage       注册信息
     */
    @Override
    public void register(SocketConnection connection, RegisterMessage registerMessage) {
        Document document = messageCollection.find(Filters.eq("target",registerMessage.getTargetName())).first();
        logger.info(" connection id is :"+connection.id());
        Document update = new Document("socket",connection.id())
                .append("target",registerMessage.getTargetName())
                .append("active",true);
        if(null==document){
            messageCollection.insertOne(update);
        }else{
            messageCollection.replaceOne(document,update);
        }
        if(null!=registerListener){
            registerListener.renderConnection(loadAllConnection());
        }
    }

    @Override
    public void closed(SocketConnection connection) {
        Document document = messageCollection.find(Filters.eq("socket",connection.id())).first();
        if(null!=document){
            document.append("active",false);
            messageCollection.replaceOne(Filters.eq("_id",document.getObjectId("_id")),document);
        }
        if(null!=closedListener){
            closedListener.renderConnection(loadAllConnection());
        }
    }

    @Override
    public String findConnectionId(String target) {
        Document document = messageCollection.find(Filters.eq("target",target)).first();
        if(null==document){
            return null;
        }else{
            return document.getString("socket");
        }
    }

    public void setMongoService(MongoService mongoService) {
        this.mongoService = mongoService;
    }

    public void setCollectionName(String collectionName) {
        this.collectionName = collectionName;
    }


    @Override
    public List<ConnectionInfo> loadAllConnection() {
        List<ConnectionInfo> connectionInfos = new ArrayList<>();
        messageCollection.find().forEach((Consumer<Document>) document -> {
            ConnectionInfo info = JSON.parseObject(document.toJson(),ConnectionInfo.class);
            connectionInfos.add(info);
        });
        return connectionInfos;
    }

    @Override
    public void bindingConnectionRegister(ConnectionRenderListener connectionRenderListener) {
        this.registerListener = connectionRenderListener;
    }

    @Override
    public void bindingConnectionClosed(ConnectionRenderListener connectionRenderListener) {
        this.closedListener = connectionRenderListener;
    }
}
